package com.rengwuxian.coursecoroutines._4_flow

import com.rengwuxian.coursecoroutines.common.Contributor
import com.rengwuxian.coursecoroutines.common.gitHub
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.receiveAsFlow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.suspendCancellableCoroutine
import retrofit2.Call
import retrofit2.Callback
import retrofit2.Response
import kotlin.coroutines.EmptyCoroutineContext
import kotlin.coroutines.cancellation.CancellationException

/**
 * Flow的构建
 * flow()、flowOf()、asFlow()、Channel.receive-/consumeAsFlow()、channelFlow()、callbackFlow()
 *
 * flowOf()是flow()的便捷函数而已，就是用flow()创建一个flow，内部逻辑对每个元素调用emit()
 * asFlow()，给它数据用来创建Flow，然后依次发送数据。是一个扩展函数，可以将各种Iterable类型的对象转换成Flow
 *     1.比如：
 *     listOf(1, 2, 3).asFlow()
 *     内部原理就是：
 *     public fun <T> Iterable<T>.asFlow(): Flow<T> = flow {
 *        forEach { value ->
 *          emit(value)
 *        }
 *      }
 *      将list中的每个原理emit出来。
 *     2.比如：
 *     sequenceOf(1, 2, 3).asFlow() // 同步的，非挂起式的，用这个sequence来创建Flow
 * 使用Channel对象来创建Flow： // Channel里发送的数据只能被接收一次。
 *     冷热的区别就是生产和接收数据是否是独立的。
 *     那上游是Channel,然后转成Flow是冷还是热？双阶段生产流程。-- 热的，延续了Channel的特性。
 *     第二阶段的Flow会把上游Channel生产的数据给掐住，等它的下游索要了再开始发送。
 *   1.Channel.receiveAsFlow() // 接收数据，是冷的，有多个collect时，数据会被瓜分。行为模式上，是热的，数据被瓜分，而不是独立。
 *     比如：
 *         val channel = Channel<Int>()
 *         val flow6 = channel.receiveAsFlow()
 *         scope.launch {
 *         flow6.collect {
 *            println("Flow6 - 1: $it")
 *          }
 *        }
 *        scope.launch {
 *          flow6.collect {
 *            println("Flow6 - 2: $it")
 *          }
 *        }
 *        channel.send(1)
 *        channel.send(2)
 *        channel.send(3)
 *        channel.send(4)
 *     输出：
 *         Flow6 - 1: 1
 *         Flow6 - 2: 2
 *         Flow6 - 1: 3
 *         Flow6 - 2: 4
 *    2.Channel.consumeAsFlow() // 被collect之后，内部就会标记为已消费，后续再被collect，就会抛异常。ReceiveChannel.consumeAsFlow can be collected just once
 *      这也是consumeAsFlow和receiveAsFlow的唯一区别。
 *      Channel提供的数据，只有一个collect才有意义，这个异常相当于一个安全提醒了。
 *      一般情况下用不到，直接用Flow就好。
 * channelFlow{}:
 *     1.是使用Channel来作为生产工具的Flow，它创建的Flow是直到collect的时候才会创建Channel，开始生产。
 *               也就是说，多次collect，会创建多个Channel，每个Channel的生产流程是互相隔离，各自独立的。
 *               从这个概念上看是冷的，延续了Flow的特性。
 *               Channel是跨协程通信的工具，可以在里面启动子协程了，
 *               比如：而同样写法的flow里是启动不了子协程的。
 *               val flow7 = channelFlow {
 *               launch {
 *                 delay(2000)
 *                 send(2)
 *               }
 *               delay(1000)
 *               send(1)
 *               }
 *     2.Flow的emit是不允许换协程的，collect在哪个协程，emit就要在哪个，如果你有跨协程生产数据的场景，用channelFlow{}
 *     3.可以和传统回调式API做对接（不是最合适的），比如：
 *           val flow10 = channelFlow {
 *               gitHub.contributorsCall("square", "retrofit")
 *                 .enqueue(object : Callback<List<Contributor>> {
 *                   override fun onResponse(call: Call<List<Contributor>>, response: Response<List<Contributor>>) {
 *                     trySend(response.body()!!)
 *                     close()
 *                   }
 *
 *                   override fun onFailure(call: Call<List<Contributor>>, error: Throwable) {
 *                     cancel(CancellationException(error))
 *                   }
 *                 })
 *               awaitClose() // 让协程不结束，防止回调还没有回来，整个协程就结束了
 *            }
 *  callbackFlow{}: 和channelFlow{}一摸一样，完全没有区别，除了以下：
 *      更适合跟回调式API做对接，内部就是channelFlow{},然后强制调用awaitClose(),
 *      如果忘记调用awaitClose()，会抛异常来提醒你。
 *      如果单次回调，用suspendCancellableCoroutine更方便，如果是多次，连续数据流用callbackFlow{}。
 *      可以看作是Flow版的suspendCancellableCoroutine。
 */
fun main() = runBlocking<Unit> {
  val flow1 = flowOf(1, 2, 3)
  val flow2 = listOf(1, 2, 3).asFlow()
  val flow3 = setOf(1, 2, 3).asFlow()
  val flow4 = sequenceOf(1, 2, 3).asFlow()
  val channel = Channel<Int>()
  val flow5 = channel.consumeAsFlow()
  val flow6 = channel.receiveAsFlow()
  val flow7 = channelFlow {
    launch {
      delay(2000)
      send(2)
    }
    delay(1000)
    send(1)
  }
  val flow8 = flow {
    launch {
      delay(2000)
      emit(2)
    }
    delay(1000)
    emit(1)
  }
  val flow9 = callbackFlow {
    gitHub.contributorsCall("square", "retrofit")
      .enqueue(object : Callback<List<Contributor>> {
        override fun onResponse(call: Call<List<Contributor>>, response: Response<List<Contributor>>) {
          trySend(response.body()!!)
          close()
        }

        override fun onFailure(call: Call<List<Contributor>>, error: Throwable) {
          cancel(CancellationException(error))
        }
      })
    awaitClose()
  }

  val flow10 = channelFlow {
    gitHub.contributorsCall("square", "retrofit")
      .enqueue(object : Callback<List<Contributor>> {
        override fun onResponse(call: Call<List<Contributor>>, response: Response<List<Contributor>>) {
          trySend(response.body()!!)
          close()
        }

        override fun onFailure(call: Call<List<Contributor>>, error: Throwable) {
          cancel(CancellationException(error))
        }
      })
    awaitClose() // 让协程不结束，防止回调还没有回来，整个协程就结束了
  }

  val scope = CoroutineScope(EmptyCoroutineContext)
  scope.launch {
    flow7.collect {
      println("channelFlow with callback: $it")
    }
    /*flow8.collect {
      println("channelFlow: $it")
    }*/
    /*flow5.collect {
      println("Flow6 - 1: $it")
    }*/
  }

/*  scope.launch {
    flow5.collect {
      println("Flow6 - 1: $it")
    }
  }
  scope.launch {
    flow5.collect {
      println("Flow6 - 2: $it")
    }
  }
  channel.send(1)
  channel.send(2)
  channel.send(3)
  channel.send(4)*/
  delay(10000)
}